package com.sunday.common.mq.rocket.brave.study.base.e12_POP;

import lombok.Data;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * https://baijiahao.baidu.com/s?id=1716353444200487221&wfr=spider&for=pc
 * <p>
 * POP/Push 模式切换方式
 * <p>
 * 可以通过两种方式进行切换。
 * <p>
 * 1、命令行
 * <p>
 * mqadmin setConsumeMode -c cluster -t topic -g group -m POP -n 8
 * 2、代码切换
 * <p>
 * public static final String CONSUMER_GROUP = "CID_JODIE_1";
 * public static final String TOPIC = "TopicTest";
 * // Or use AdminTools directly: mqadmin setConsumeMode -c cluster -t topic -g group -m POP -n 8
 * private static void switchPop() throws Exception {
 * DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();
 * mqAdminExt.start();
 * ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
 * Set<String> brokerAddrs = clusterInfo.getBrokerAddrTable().values().stream().map(BrokerData::selectBrokerAddr).collect(Collectors.toSet());
 * for (String brokerAddr : brokerAddrs) {
 * mqAdminExt.setMessageRequestMode(brokerAddr, TOPIC, CONSUMER_GROUP, MessageRequestMode.POP, 8, 3_000);
 * }
 * }
 * 通过下面 POP Consumer Demo，我们看到 POP Consumer 跟 Push API 基本是统一，使用也比较简单，相比 Push API 只是多了一步消费模式切换。
 * <p>
 * <p>
 * Push & POP Retry 队列差异
 * <p>
 * 在使用 POP 消费模式时我们只需要在 Push API 的基础上切换模式即可，对于 Broker 来说还是需要做一些处理的。主要需要处理的地方是 retry 队列。
 * <p>
 * Push 和 POP 模式对 retry 队列处理不一样
 * <p>
 * Push 的 retry 处理
 * 服务端有一个 %RETRY%ConsumerGroup 队列
 * 客户端会有拉取任务拉取这个队列的消息。
 * POP 的 retry 处理
 * 服务端针对每个Topic，都有一个名为 %RETRY%ConsumerGroup_Topic 的 retry 队列
 * 客户端没有专门针对 retry 队列的拉任务，每次普通 POP 请求都有一定概率消费相应的 retry 队列
 * 模式切换之后，老模式的 retry 里的消息还需要继续处理，否则就丢消息了。
 * <p>
 * Push & POP 切换
 * <p>
 * Push 切换到 POP
 * <p>
 * 正常队列切换到 POP 模式
 * 正常队列的 POP 请求会处理对应的 POP retry 队列
 * 针对 Push retry 队列，我们保留原来 Push retry 队列的拉取任务，并且是工作在 Push 模式。
 * POP 切换到 Push
 * <p>
 * 正常队列切换到 Push 模式
 * Push retry 队列自然有相应的拉取任务
 * 之前 POP 的 retry 队列，我们在客户端自动创建拉取任务，以Push 模式去拉取。注意这里的拉取任务只拉取 POP 的 retry 队列。
 * 总结下来就是，对于 retry 队列，我们会特殊处理不参与模式切换。
 * <p>
 * 总结
 * 最后我们总结下 POP Consumer。POP 作为一种全新的消费模式，解决了 Push 模式的一些痛点，使客户端无状态，更加轻量，消费逻辑也基本都收敛到了 Broker，对多语言的支持十分的友好。在 API 层面也与 Push 完成了融合，继承了 Push API 的简单易用，同时实现了 Push，POP 之间的自由切换。
 */
@Data
@Slf4j
public class PushPopConsumerThread extends Thread {

    private DefaultMQPushConsumer consumer;

    private String node;

    private AtomicInteger count = new AtomicInteger();

    public PushPopConsumerThread(String consumerGroup, String namesrvAddr, String topic, String node) throws Exception {
        this.node = node;
        this.consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
//        consumer.setConsumeThreadMin(1);
//        consumer.setConsumeThreadMax(1);
        consumer.subscribe(topic, "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
//            log.info("Receive New Messages: {}", msgs.size());
            msgs.stream().forEach(messageExt -> log.info("{} -> {} -> {}", this.node, new String(messageExt.getBody()), messageExt));
            count.incrementAndGet();
            // 标记该消息已经被成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 强制使用客户端再平衡 这个值非常重要，否则push模式下pop将无法唤醒使用
        consumer.setClientRebalance(false);
    }

    @Override
    public void run() {
        try {
            consumer.start();
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        log.info("Consumer Started. {}", node);
    }

}
